package com.sl.core.engine.thread;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.BooleanUtil;
import com.sl.core.engine.step.impl.AbstractProcessComp;
import com.sl.core.engine.trans.ProcessInstance;
import org.apache.hop.core.Const;
import org.apache.hop.core.logging.HopLogStore;
import org.apache.hop.core.logging.ILoggingObject;
import org.apache.hop.core.logging.LoggingRegistry;
import org.apache.hop.core.logging.Metrics;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.RunThread;
import org.apache.hop.pipeline.transform.TransformMetaDataCombi;

import java.util.List;

public class ProcessRunSingleThreaded extends RunThread {
    private static Class<?> PKG = BaseTransform.class;

    private List<TransformMetaDataCombi> steps;

    private PipelineMeta transMeta;

    private Pipeline trans;

    public ProcessRunSingleThreaded(List<TransformMetaDataCombi> combi, PipelineMeta pipelineMeta, Pipeline pipeline) {
        super(CollUtil.get(combi, 0));
        this.steps = combi;
        this.transMeta = pipelineMeta;
        this.trans = pipeline;
    }

    @Override
    public void run() {
        try {

            steps.sort((c1, c2) -> {

                boolean c1BeforeC2 = transMeta.findPrevious(c2.transformMeta, c1.transformMeta);
                if (c1BeforeC2) {
                    return -1;
                } else {
                    return 1;
                }
            });

            boolean[] stepDone = new boolean[steps.size()];
            int nrDone = 0;
            while (nrDone < steps.size() && !trans.isStopped()) {
                for (int i = 0; i < steps.size() && !trans.isStopped(); i++) {
                    TransformMetaDataCombi combi = steps.get(i);
                    if (!stepDone[i]) {
                        boolean cont = false;
                        try {
                            cont = combi.transform.processRow();
                        } catch (Throwable t) {
                            if (combi.transform instanceof AbstractProcessComp comp) {
                                comp.addThrowable(t);
                                comp.setErrors(comp.getErrors() + 1);
                            }

                            if (trans instanceof ProcessInstance processInstance) {
                                processInstance.callStepStatus(combi.transform);
                                Boolean fastFailFlag = processInstance.getFastFailFlag();
                                if (BooleanUtil.isTrue(fastFailFlag)) {
                                    throw t;
                                }
                            }
                        }
                        if (!cont) {
                            if (trans instanceof ProcessInstance processInstance) {
                                processInstance.callStepStatus(combi.transform);
                            }
                            stepDone[i] = true;
                            nrDone++;
                        }
                    }
                }
            }
        } catch (Throwable t) {

            try {
                // check for OOME
                if (t instanceof OutOfMemoryError) {
                    // Handle this different with as less overhead as possible to get an error message in the log.
                    // Otherwise it crashes likely with another OOME in Me$$ages.getString() and does not log
                    // nor call the setErrors() and stopAll() below.
                    trans.getLogChannel().logError("UnexpectedError: ", t);
                } else {
                    t.printStackTrace();
                    trans.getLogChannel().logError(BaseMessages.getString("System.Log.UnexpectedError"), t);
                }

                String logChannelId = trans.getLogChannel().getLogChannelId();
                ILoggingObject loggingObject = LoggingRegistry.getInstance().getLoggingObject(logChannelId);
                String parentLogChannelId = loggingObject.getParent().getLogChannelId();
                List<String> logChannelChildren = LoggingRegistry.getInstance().getLogChannelChildren(parentLogChannelId);
                int childIndex = Const.indexOfString(trans.getLogChannel().getLogChannelId(), logChannelChildren);
                if (trans.getLogChannel().isDebug()) {
                    trans.getLogChannel().logDebug("child index = " + childIndex + ", logging object : " + loggingObject.toString() + " parent=" + parentLogChannelId);
                }
                HopLogStore.getAppender().getBuffer("2bcc6b3f-c660-4a8b-8b17-89e8cbd5b29b", false);
                // baseStep.logError(Const.getStackTracker(t));
            } catch (OutOfMemoryError e) {
                e.printStackTrace();
            }
        } finally {
            for (TransformMetaDataCombi combi : steps) {
                combi.transform.dispose();
                combi.transform.getLogChannel().snap(Metrics.METRIC_TRANSFORM_EXECUTION_STOP);
                try {
                    long li = combi.transform.getLinesInput();
                    long lo = combi.transform.getLinesOutput();
                    long lr = combi.transform.getLinesRead();
                    long lw = combi.transform.getLinesWritten();
                    long lu = combi.transform.getLinesUpdated();
                    long lj = combi.transform.getLinesRejected();
                    long e = combi.transform.getErrors();
                    if (li > 0 || lo > 0 || lr > 0 || lw > 0 || lu > 0 || lj > 0 || e > 0) {
                        combi.transform.getLogChannel().logBasic(BaseMessages.getString(PKG, "BaseStep.Log.SummaryInfo", String.valueOf(li),
                                String.valueOf(lo), String.valueOf(lr), String.valueOf(lw),
                                String.valueOf(lu), String.valueOf(e + lj)));
                    } else {
                        combi.transform.getLogChannel().logDetailed(BaseMessages.getString(PKG, "BaseStep.Log.SummaryInfo", String.valueOf(li),
                                String.valueOf(lo), String.valueOf(lr), String.valueOf(lw),
                                String.valueOf(lu), String.valueOf(e + lj)));
                    }
                } catch (Throwable t) {
                    //
                    // it's likely an OOME, so we don't want to introduce overhead by using BaseMessages.getString(), see above
                    //
                    combi.transform.getLogChannel().logError("UnexpectedError: " + Const.getStackTracker(t));
                } finally {
                    combi.transform.markStop();
                }

            }
        }
    }
}
